package com.zhang.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zhang.gmall.realtime.bean.ProvinceStats;
import com.zhang.gmall.realtime.utils.ClickHouseUtil;
import com.zhang.gmall.realtime.utils.DateTimeUtil;
import com.zhang.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;


import java.time.Duration;
import java.util.Date;
import java.util.HashSet;

/**
 * @title:
 * @author: zhang
 * @date: 2022/1/9 21:02
 */
public class ProvinceStatsApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //1.1 开启CK并指定状态后端为FS    memory  fs  rocksdb
      /*  env.setStateBackend(new FsStateBackend("hdfs://hadoop302:8020/gmall-flink/ck"));
        env.enableCheckpointing(5000L);
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(10000L);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000);
        System.setProperty("HADOOP_USER_NAME", "zhang");*/

        //TODO 2.读取kafka 主题的数据创建流
        String groupId = "province_stats";
        String orderWideTopic = "dwm_order_wide";

        SingleOutputStreamOperator<ProvinceStats> provinceStatsDS = env.addSource(MyKafkaUtil.getKafkaConsumer(orderWideTopic, groupId))
                .map(line -> {
                    JSONObject jsonObject = JSON.parseObject(line);
                    HashSet<Long> orderIds = new HashSet<>();
                    orderIds.add(jsonObject.getLong("order_id"));
                    return ProvinceStats.builder()
                            .province_id(jsonObject.getLong("province_id"))
                            .province_name(jsonObject.getString("province_name"))
                            .province_area_code(jsonObject.getString("province_area_code"))
                            .province_iso_code(jsonObject.getString("province_iso_code"))
                            .province_3166_2_code(jsonObject.getString("province_3166_2_code"))
                            .order_amount(jsonObject.getBigDecimal("split_total_amount"))
                            .orderIdSet(orderIds)
                            .ts(DateTimeUtil.toTs(jsonObject.getString("create_time")))
                            .build();
                });

        //TODO 3.提取时间戳生成WaterMark
        SingleOutputStreamOperator<ProvinceStats> provinceStatsWithWMDS = provinceStatsDS.assignTimestampsAndWatermarks(WatermarkStrategy.<ProvinceStats>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                .withTimestampAssigner(new SerializableTimestampAssigner<ProvinceStats>() {
                    @Override
                    public long extractTimestamp(ProvinceStats element, long recordTimestamp) {
                        return element.getTs();
                    }
                }));

        //TODO 4.按照维度分组
        KeyedStream<ProvinceStats, Tuple5<String, String, String, String, String>> provinceStatsKeyedStream = provinceStatsWithWMDS.keyBy(new KeySelector<ProvinceStats, Tuple5<String, String, String, String, String>>() {
            @Override
            public Tuple5<String, String, String, String, String> getKey(ProvinceStats provinceStats) throws Exception {
                return new Tuple5<String, String, String, String, String>(
                        String.valueOf(provinceStats.getProvince_id()),
                        provinceStats.getProvince_name(),
                        provinceStats.getProvince_area_code(),
                        provinceStats.getProvince_iso_code(),
                        provinceStats.getProvince_3166_2_code()
                );
            }
        });

        //TODO 5.开窗
        SingleOutputStreamOperator<ProvinceStats> provinceStatsDataStream = provinceStatsKeyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<ProvinceStats>() {
                    @Override
                    public ProvinceStats reduce(ProvinceStats value1, ProvinceStats value2) throws Exception {
                        value1.setOrder_amount(value1.getOrder_amount().add(value2.getOrder_amount()));
                        return value1;
                    }
                }, new WindowFunction<ProvinceStats, ProvinceStats, Tuple5<String, String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple5<String, String, String, String, String> stringStringStringStringStringTuple5, TimeWindow timeWindow, Iterable<ProvinceStats> iterable, Collector<ProvinceStats> collector) throws Exception {

                        //取出数据
                        ProvinceStats provinceStats = iterable.iterator().next();

                        //设置窗口时间
                        provinceStats.setStt(DateTimeUtil.toYMDhms(new Date(timeWindow.getStart())));
                        provinceStats.setEdt(DateTimeUtil.toYMDhms(new Date(timeWindow.getEnd())));

                        //设置订单数
                        provinceStats.setOrder_count((long) provinceStats.getOrderIdSet().size());

                        //将数据写出
                        collector.collect(provinceStats);
                    }
                });
        //TODO 6.写入Clickhouse
        //打印测试
        provinceStatsDataStream.print(">>>>>>>>");

        provinceStatsDataStream.addSink(ClickHouseUtil.getSink("insert into province_stats values(?,?,?,?,?,?,?,?,?,?)"));


        //TODO 7.启动任务
        env.execute("ProvinceStatsApp");
    }
}
